/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.remoting.netty;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.remoting.ChannelEventListener;
import com.alibaba.rocketmq.remoting.InvokeCallback;
import com.alibaba.rocketmq.remoting.RPCHook;
import com.alibaba.rocketmq.remoting.common.Pair;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.common.SemaphoreReleaseOnlyOnce;
import com.alibaba.rocketmq.remoting.common.ServiceThread;
import com.alibaba.rocketmq.remoting.exception.RemotingSendRequestException;
import com.alibaba.rocketmq.remoting.exception.RemotingTimeoutException;
import com.alibaba.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;
import com.alibaba.rocketmq.remoting.protocol.RemotingSysResponseCode;

/**
 * Server与Client公用抽象类
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-13
 */
public abstract class NettyRemotingAbstract {
	private static final Logger plog = LoggerFactory.getLogger(RemotingHelper.RemotingLogName);

	// 信号量，Oneway情况会使用，防止本地Netty缓存请求过多
	protected final Semaphore semaphoreOneway;

	// 信号量，异步调用情况会使用，防止本地Netty缓存请求过多
	protected final Semaphore semaphoreAsync;

	// 缓存所有对外请求
	protected final ConcurrentHashMap<Integer /* opaque */, ResponseFuture> responseTable = new ConcurrentHashMap<Integer, ResponseFuture>(256);

	// 默认请求代码处理器
	protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessor;

	// 注册的各个RPC处理器
	protected final HashMap<Integer/* request code */, Pair<NettyRequestProcessor, ExecutorService>> processorTable = new HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>>(64);

	protected final NettyEventExecuter nettyEventExecuter = new NettyEventExecuter();

	public abstract ChannelEventListener getChannelEventListener();

	public abstract RPCHook getRPCHook();

	public void putNettyEvent(final NettyEvent event) {
		this.nettyEventExecuter.putNettyEvent(event);
	}

	class NettyEventExecuter extends ServiceThread {
		private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<NettyEvent>();
		private final int MaxSize = 10000;

		public void putNettyEvent(final NettyEvent event) {
			if (this.eventQueue.size() <= MaxSize) {
				this.eventQueue.add(event);
			} else {
				plog.warn("event queue size[{}] enough, so drop this event {}", this.eventQueue.size(), event.toString());
			}
		}

		@Override
		public void run() {
			plog.info(this.getServiceName() + " service started");

			final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();

			while (!this.isStoped()) {
				try {
					NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
					if (event != null && listener != null) {
						switch (event.getType()) {
							case IDLE :
								listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
								break;
							case CLOSE :
								listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
								break;
							case CONNECT :
								listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
								break;
							case EXCEPTION :
								listener.onChannelException(event.getRemoteAddr(), event.getChannel());
								break;
							default :
								break;

						}
					}
				} catch (Exception e) {
					plog.warn(this.getServiceName() + " service has exception. ", e);
				}
			}

			plog.info(this.getServiceName() + " service end");
		}

		@Override
		public String getServiceName() {
			return NettyEventExecuter.class.getSimpleName();
		}
	}

	public NettyRemotingAbstract(final int permitsOneway, final int permitsAsync) {
		this.semaphoreOneway = new Semaphore(permitsOneway, true);
		this.semaphoreAsync = new Semaphore(permitsAsync, true);
	}

	public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
		final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
		final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;

		if (pair != null) {
			Runnable run = new Runnable() {
				@Override
				public void run() {
					try {
						RPCHook rpcHook = NettyRemotingAbstract.this.getRPCHook();
						if (rpcHook != null) {
							rpcHook.doBeforeRequest(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
						}
						// 根据NettyRequestProcessor的类型执行相应的processRequest方法
						// brokerController中registerProcessor方法中注册了各种NettyRequestProcessor
						// 如：SendMessageProcessor、PullMessageProcessor、NettyRequestProcessor、ClientManageProcessor、AdminBrokerProcessor
						final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
						if (rpcHook != null) {
							rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
						}

						if (!cmd.isOnewayRPC()) {
							if (response != null) {
								response.setOpaque(cmd.getOpaque());
								response.markResponseType();
								try {
									ctx.writeAndFlush(response);
								} catch (Throwable e) {
									plog.error("process request over, but response failed", e);
									plog.error(cmd.toString());
									plog.error(response.toString());
								}
							} else {
								// 收到请求，但是没有返回应答，可能是processRequest中进行了应答，忽略这种情况
							}
						}
					} catch (Throwable e) {
						plog.error("process request exception", e);
						plog.error(cmd.toString());

						if (!cmd.isOnewayRPC()) {
							final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR, //
									RemotingHelper.exceptionSimpleDesc(e));
							response.setOpaque(cmd.getOpaque());
							ctx.writeAndFlush(response);
						}
					}
				}
			};

			try {
				// 这里需要做流控，要求线程池对应的队列必须是有大小限制的
				pair.getObject2().submit(run);
			} catch (RejectedExecutionException e) {
				// 每个线程10s打印一次
				if ((System.currentTimeMillis() % 10000) == 0) {
					plog.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) //
							+ ", too many requests and system thread pool busy, RejectedExecutionException " //
							+ pair.getObject2().toString() //
							+ " request code: " + cmd.getCode());
				}

				if (!cmd.isOnewayRPC()) {
					final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY, "too many requests and system thread pool busy, please try another server");
					response.setOpaque(cmd.getOpaque());
					ctx.writeAndFlush(response);
				}
			}
		} else {
			String error = " request type " + cmd.getCode() + " not supported";
			final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
			response.setOpaque(cmd.getOpaque());
			ctx.writeAndFlush(response);
			plog.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
		}
	}

	public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
		// 1、根据返回的响应对象RemotingCommand的opaque（请求序列号）从NettyRemotingAbstract.responseTable: ConcurrentHashMap<Integer /* opaque */, ResponseFuture>变量中获取ResponseFuture对象（在发送之前存入的该对象）
		final ResponseFuture responseFuture = responseTable.get(cmd.getOpaque());
		if (responseFuture != null) {
			// 3、若该ResponseFuture对象不为null，
			// 将响应对象RemotingCommand赋值给ResponseFuture.responseCommand变量；
			responseFuture.setResponseCommand(cmd);

			responseFuture.release();

			// 4、若ResponseFuture.invokeCallback:InvokeCallback变量不为空（在异步发送的情况下该变量不为空）
			if (responseFuture.getInvokeCallback() != null) {
				boolean runInThisThread = false;
				// 首先获取NettyRemotingClient.publicExecutor线程池
				ExecutorService executor = this.getCallbackExecutor();
				if (executor != null) {
					try {
						// 若存在该线程池则初始化Runnable匿名线程，将该匿名线程提交到线程池中；
						executor.submit(new Runnable() {
							@Override
							public void run() {
								try {
									responseFuture.executeInvokeCallback();
								} catch (Throwable e) {
									plog.warn("excute callback in executor exception, and callback throw", e);
								}
							}
						});
					} catch (Exception e) {
						runInThisThread = true;
						plog.warn("excute callback in executor exception, maybe executor busy", e);
					}
				} else {
					runInThisThread = true;
				}

				if (runInThisThread) {
					try {
						// 若没有该线程池则直接在主线程中调用ResponseFuture.executeInvokeCallback()方法
						responseFuture.executeInvokeCallback();
					} catch (Throwable e) {
						plog.warn("executeInvokeCallback Exception", e);
					}
				}
			} else {
				// 5、若ResponseFuture.invokeCallback:InvokeCallback变量为空（在同步方式拉取消息的情况下），
				// 则调用ResponseFuture.putResponse (RemotingCommand responseCommand)方法
				// 首先将响应对象RemotingCommand赋值给ResponseFuture.responseCommand变量，
				// 然后唤醒ResponseFuture.waitResponse方法的等待；
				responseFuture.putResponse(cmd);
			}
		} else {
			// 2、若该ResponseFuture对象为null，则啥也不干，就打警告日志
			plog.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
			plog.warn(cmd.toString());
		}

		responseTable.remove(cmd.getOpaque());
	}

	public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
		final RemotingCommand cmd = msg;
		if (cmd != null) {
			switch (cmd.getType()) {
				case REQUEST_COMMAND :
					processRequestCommand(ctx, cmd);
					break;
				case RESPONSE_COMMAND :
					processResponseCommand(ctx, cmd);
					break;
				default :
					break;
			}
		}
	}

	abstract public ExecutorService getCallbackExecutor();

	public void scanResponseTable() {
		Iterator<Entry<Integer, ResponseFuture>> it = this.responseTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<Integer, ResponseFuture> next = it.next();
			ResponseFuture rep = next.getValue();
			// 遍历该列表中的ResponseFuture对象
			// 检查等待响应是否超时
			if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
				// 将该对象从responseTable列表中删除；
				it.remove();
				try {
					// 调用ResponseFuture. executeInvokeCallback()方法，
					rep.executeInvokeCallback();
				} catch (Throwable e) {
					plog.warn("scanResponseTable, operationComplete Exception", e);
				} finally {
					rep.release();
				}

				plog.warn("remove timeout request, " + rep);
			}
		}
	}

	public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis) throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
		try {
			// A）使用请求的序列号（opaue）、超时时间初始化ResponseFuture对象
			final ResponseFuture responseFuture = new ResponseFuture(request.getOpaque(), timeoutMillis, null, null);
			// 以请求消息的序列号为key值，存入NettyRemotingAbstract.responseTable:ConcurrentHashMap<Integer /* opaque */, ResponseFuture>变量中
			this.responseTable.put(request.getOpaque(), responseFuture);

			// B）调用Channel.writeAndFlush(Object msg)方法将请求对象RemotingCommand发送给Broker；
			// 然后调用addListener(GenericFutureListener<? extends Future<? super Void>> listener)方法添加内部匿名类：
			// 该内部匿名类实现了ChannelFutureListener接口的operationComplete方法，在发送完成之后回调该监听类的operationComplete方法，在该方法中，
			channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture f) throws Exception {
					// 检查是否发送成功
					if (f.isSuccess()) {
						// 若成功则置ResponseFuture对象的sendRequestOK等于true并
						responseFuture.setSendRequestOK(true);
						// 退出此回调方法等待响应结果
						return;
					} else {
						// 若不成功则置ResponseFuture对象的sendRequestOK等于false
						responseFuture.setSendRequestOK(false);
					}
					// 发送不成功
					// 从NettyRemotingAbstract.responseTable中删除此请求序列号（opaue）的记录
					responseTable.remove(request.getOpaque());
					responseFuture.setCause(f.cause());
					responseFuture.putResponse(null);
					plog.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
					plog.warn(request.toString());
				}
			});

			// 若是同步发送会唤醒等待响应的ResponseFuture.waitResponse方法；
			// C）调用ResponseFuture.waitResponse(long timeoutMillis)方法等待响应结果；在发送失败或者收到响应消息（详见5.10.3小节）或者超时的情况下会唤醒该方法返回ResponseFuture.responseCommand变量值；
			RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
			if (null == responseCommand) {
				// D）若上一步返回的responseCommand值为null
				if (responseFuture.isSendRequestOK()) {
					// 若ResponseFuture.sendRequestOK为true，则抛出RemotingTimeoutException异常
					throw new RemotingTimeoutException(RemotingHelper.parseChannelRemoteAddr(channel), timeoutMillis, responseFuture.getCause());
				} else {
					throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), responseFuture.getCause());
				}
			}

			// E）若上一步返回的responseCommand值不为null，则返回responseCommand变量值；
			return responseCommand;
		} finally {
			this.responseTable.remove(request.getOpaque());
		}
	}

	public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis, final InvokeCallback invokeCallback) throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
		// A）利用java.util.concurrent.Semaphore.tryAcquire(long timeout,TimeUnitunit)获取信号量，
		// 保证该方法的业务逻辑同时执行的线程个数；
		boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
		if (acquired) {
			// 用Semaphore初始化的SemaphoreReleaseOnlyOnce对象（该对象是确保在释放信号量是只释放一次）
			final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);

			// B)初始化ResponseFuture对象，其中InvokeCallback匿名类就是该对象的InvokeCallback变量值
			final ResponseFuture responseFuture = new ResponseFuture(request.getOpaque(), timeoutMillis, invokeCallback, once);
			// 根据请求的序列号（opaue）作为key值，将该ResponseFuture对象存入NettyRemotingAbstract.ResponseTable: ConcurrentHashMap<Integer /* opaque */, ResponseFuture>变量中；
			this.responseTable.put(request.getOpaque(), responseFuture);
			try {
				// C）调用Channel.writeAndFlush(Object msg)方法将请求对象发送给Broker，并且添加监听器，再消息发送完成之后回调该监听器，该监听器是内部匿名类，该类实现了ChannelFutureListener接口的operationComplete(ChannelFuture f)方法
				channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
					@Override
					public void operationComplete(ChannelFuture f) throws Exception {
						// C.1）首先调用ChannelFuture.isSuccess()方法检查是否发送成功，
						if (f.isSuccess()) {
							// 若成功则置ResponseFuture对象的sendRequestOK等于true并退出此回调方法等待对方的响应消息；
							responseFuture.setSendRequestOK(true);
							return;
						} else {
							// 若不成功则置ResponseFuture对象的sendRequestOK等于false，然后继续执行下面的逻辑
							// 主要目的是立即向应用层返回发送失败的响应消息，无需再等待对方的响应结果；
							responseFuture.setSendRequestOK(false);
						}

						// C.3）将ResponseFuture.responseCommand变量置为null；
						responseFuture.putResponse(null);
						// C.2）在收到响应消息之后以响应ID（即请求ID）从NettyRemotingAbstract. ResponseTable变量中取ResponseFuture对象
						responseTable.remove(request.getOpaque());
						try {
							// 异步发送调用InvokeCallback类的operationComplete方法，完成回调工作
							/**
							 * C.4）调用ResponseFuture.executeInvokeCallback()方法，
							 * 在该方法中执行InvokeCallback.OperationComplete(ResponseFuture)方法完成回调工作，
							 * 在executeInvokeCallback方法之前，确保ResponseFuture. executeCallbackOnlyOnce的值为false并且成功更新为true，
							 * 由于executeCallbackOnlyOnce在初始化时为false，若更新失败说明该回调方法已经执行过了，故不在执行；
							 */
							responseFuture.executeInvokeCallback();
						} catch (Throwable e) {
							plog.warn("excute callback in writeAndFlush addListener, and callback throw", e);
						} finally {
							// C.5）最后调用SemaphoreReleaseOnlyOnce对象的realse，释放信号量；
							responseFuture.release();
						}

						plog.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
						plog.warn(request.toString());
					}
				});
			} catch (Exception e) {
				responseFuture.release();
				plog.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
				throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
			}
		} else {
			if (timeoutMillis <= 0) {
				throw new RemotingTooMuchRequestException("invokeAsyncImpl invoke too fast");
			} else {
				String info = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d", //
						timeoutMillis, //
						this.semaphoreAsync.getQueueLength(), //
						this.semaphoreAsync.availablePermits()//
				);
				plog.warn(info);
				plog.warn(request.toString());
				throw new RemotingTimeoutException(info);
			}
		}
	}

	public void invokeOnewayImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis) throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
		request.markOnewayRPC();
		boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
		if (acquired) {
			final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
			try {
				channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
					@Override
					public void operationComplete(ChannelFuture f) throws Exception {
						once.release();
						if (!f.isSuccess()) {
							plog.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
							plog.warn(request.toString());
						}
					}
				});
			} catch (Exception e) {
				once.release();
				plog.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
				throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
			}
		} else {
			if (timeoutMillis <= 0) {
				throw new RemotingTooMuchRequestException("invokeOnewayImpl invoke too fast");
			} else {
				String info = String.format("invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d", //
						timeoutMillis, //
						this.semaphoreAsync.getQueueLength(), //
						this.semaphoreAsync.availablePermits()//
				);
				plog.warn(info);
				plog.warn(request.toString());
				throw new RemotingTimeoutException(info);
			}
		}
	}
}
